#include "wtk/core/wtk_alloc.h"
#include "wtk/os/wtk_broadcast_msg_queue.h"

int wtk_broadcast_msg_queue_init(wtk_broadcast_msg_queue_t *q, int num, wtk_broadcast_msg_queue_gabbage_collector collector, void * data)
{
	int i;
	wtk_queue_init(&(q->q_msg));
	q->receiver_num = num<=0?1:num;
	q->collector = collector;
	q->collector_data = data;

    pthread_mutex_init(&(q->msg_mutex), 0);
    pthread_cond_init(&(q->cond), NULL);

	q->info = (wtk_broadcast_msg_queue_reader_info_t *)wtk_calloc(sizeof(wtk_broadcast_msg_queue_reader_info_t), num);
	for(i = 0; i < num; ++i)
	{
		wtk_broadcast_msg_queue_reader_info_init(q->info + i);
	}
    return 0;
}

int wtk_broadcast_msg_queue_clear(wtk_broadcast_msg_queue_t *q)
{
    int i;
    int num =  q->receiver_num;
	wtk_queue_node_t* qn;

	wtk_broadcast_msg_queue_sweep_gabbage(q);

    for(i = 0; i < num; ++i)
	{
		wtk_broadcast_msg_queue_reader_info_clear(q->info + i);
	}

	pthread_cond_destroy(&(q->cond));

    //clear the remaining queue nodes
    pthread_mutex_lock(&(q->msg_mutex));
    while(qn = wtk_queue_pop(&(q->q_msg)))
    {
        pthread_mutex_unlock(&(q->msg_mutex));
        q->collector(q->collector_data, qn);
        pthread_mutex_lock(&(q->msg_mutex));
    }
    pthread_mutex_unlock(&(q->msg_mutex));

	pthread_mutex_destroy(&(q->msg_mutex));

	wtk_queue_init(&(q->q_msg));

    free(q->info);
    return 0;
}

int wtk_broadcast_msg_queue_broadcast(wtk_broadcast_msg_queue_t *q, wtk_queue_node_t *n)
{
	pthread_mutex_lock(&(q->msg_mutex));
	wtk_queue_push(&(q->q_msg), n);
	pthread_cond_broadcast(&(q->cond));
	pthread_mutex_unlock(&(q->msg_mutex));
    return 0;
}

wtk_queue_node_t* wtk_broadcast_msg_queue_get_msg(wtk_broadcast_msg_queue_t *q, int i)
{
	wtk_queue_node_t * cur_qn = q->info[i].p_cur_node;
	pthread_mutex_lock(&(q->msg_mutex));

	if(( cur_qn == NULL && q->q_msg.pop == NULL) || (cur_qn && cur_qn->next == NULL))
		pthread_cond_wait(&(q->cond), &(q->msg_mutex));

	pthread_mutex_unlock(&(q->msg_mutex));
	if(cur_qn){
		cur_qn = cur_qn->next;
        q->info[i].not_first = 1;
	}else{
		cur_qn = q->q_msg.pop;
        q->info[i].not_first = 0;
    }

	q->info[i].p_cur_node = cur_qn;
	return cur_qn;
}

int wtk_broadcast_msg_queue_return_msg(wtk_broadcast_msg_queue_t * q, int i)
{
	wtk_queue_node_t * cur_qn = q->info[i].p_cur_node;
	if(cur_qn != NULL && q->info[i].not_first)
	{
        sem_post(&(q->info[i].sem));
	}
    return 0;
}

int wtk_broadcast_msg_queue_sweep_gabbage(wtk_broadcast_msg_queue_t * q)
{
	int i, j, ret = 0;
    int svalue, min;
    int max = 0;
	int num = q->receiver_num;
	wtk_queue_node_t* qn;

    sem_getvalue(&(q->info[0].sem), &min);
    min = min <= 0? 0:min;

    for(i = 1; i < num; ++i)
    {
        sem_getvalue(&(q->info[i].sem), &svalue);
    
        if(svalue <= 0)
        {
            min = 0;
            break;
        }
        min= (min > svalue)?svalue:min;
    }

    for(j = 0; j < min; ++j)
    {
        for(i = 0; i < num; ++i)
        {
            if(sem_trywait(&(q->info[i].sem)))
            {
                wtk_debug("broadcast queue error");
                ret = 1;
                goto end;
            }
        }
        pthread_mutex_lock(&(q->msg_mutex));
        qn = wtk_queue_pop(&(q->q_msg));
        pthread_mutex_unlock(&(q->msg_mutex));
        q->collector(q->collector_data, qn);
    }
end:
    return ret;
}


int wtk_broadcast_msg_queue_reset(wtk_broadcast_msg_queue_t * q)
{
    int num = q->receiver_num;
    wtk_broadcast_msg_queue_gabbage_collector collector = q->collector;
    void * data = q->collector_data;
    wtk_broadcast_msg_queue_clear(q);
    return wtk_broadcast_msg_queue_init(q, num, collector, data);
}

